// Copyright 2012 Anno Langen. All Rights Reserved.

package org.anno;

import com.google.inject.Guice;
import org.anno.Ratings.ProcessorFactory;

import static org.anno.Ratings.USER_SHARD_COUNT;
import org.anno.Ratings.UserProcessor;

/**
 * Determines count of users and ratings in each user shard.
 */
class UserShardAnalyzer implements ProcessorFactory<UserProcessor> {

  int[] shardUserCount = new int[USER_SHARD_COUNT];
  int[] shardRatingCount = new int[USER_SHARD_COUNT];
  int maxId;
  int maxCount;

  public static void main(String[] args) throws InterruptedException {
    Ratings ratings = Guice.createInjector(Ratings.getModule()).getInstance(Ratings.class);
    UserShardAnalyzer counter = new UserShardAnalyzer();
    ratings.sendByUser(counter);
    System.out.println(counter);
    Parallel.EXECUTOR.shutdown();
  }

  public String toString() {
    StringBuilder buf = new StringBuilder();
    int shardUserMax = 0;
    int shardRatingMax = 0;
    for (int i = 0; i < shardUserCount.length; i++) {
      buf.append("Shard ").append(i).append(" users ").append(shardUserCount[i]).append(" ratings ")
          .append(shardRatingCount[i]).append("\n");
      if (shardRatingCount[i] > shardRatingMax) {
        shardRatingMax = shardRatingCount[i];
      }
      if (shardUserCount[i] > shardUserMax) {
        shardUserMax = shardUserCount[i];
      }
    }
    buf.append("Max ").append(maxCount).append(" id ").append(maxId).append(" shardUserMax ")
        .append(shardUserMax).append(" shardRatingMax ").append(shardRatingMax).append("\n");
    return buf.toString();
  }

  public UserProcessor get(final int id) {
    shardUserCount[id >> 16] += 1;
    return new UserProcessor() {
      int count;

      public void close() {
        shardRatingCount[id >> 16] += count;
        synchronized (UserShardAnalyzer.this) {
          if (count > maxCount) {
            maxId = id;
            maxCount = count;
          }
        }
      }

      public void movieScore(int movie, int score) {
        count++;
      }
    };
  }
}
